package com.dongdongshop.mq;

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;

import javax.annotation.Resource;

@RocketMQTransactionListener
public class MQCallbackTransactionListener implements RocketMQLocalTransactionListener {

    @Resource
    private RedisTemplate redisTemplate;

    /**
     * 1、MQ服务，成功接到事物消息后，执行本方法；
     * 2、处理本地事物，并将本地事物处理结果返回给MQ服务
     *
     * @return COMMIT, ROLLBACK, UNKNOWN;
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        String messageId = message.getHeaders().getId().toString();
        //将本地事物执行状态return到MQ服务，并且保存到redis备份（网络延迟，MQ服务产生回调的时候 可以从redis中获取）
        try {
            //业务处理
            System.out.println(new String((byte[]) message.getPayload()));
        } catch (Exception e) {
            //处理异常返回ROLLBACK
            redisTemplate.boundValueOps(messageId).set("ERROR");
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        //处理成功返回COMMIT
        redisTemplate.boundValueOps(messageId).set("SUCCESS");
        return RocketMQLocalTransactionState.COMMIT;
    }


    /**
     * MQ服务 由于网络等原因 未收到 本地事物处理结果，回查本地事物处理结果
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        String messageId = message.getHeaders().getId().toString();
        String status = (String) redisTemplate.boundValueOps(messageId).get();
        if ("SUCCESS".equals(status)){
            return RocketMQLocalTransactionState.COMMIT;
        }else if ("ERROR".equals(status)){
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        return RocketMQLocalTransactionState.UNKNOWN;
    }

}
